package com.jetpackframework.rxjetpack.observable;

import androidx.annotation.NonNull;

import com.jetpackframework.rxjetpack.DisposableHelper;
import com.jetpackframework.rxjetpack.Dispose;
import com.jetpackframework.rxjetpack.DisposeImpl;
import com.jetpackframework.rxjetpack.Observable;
import com.jetpackframework.rxjetpack.Observer;
import com.jetpackframework.rxjetpack.thread.Schedule;

public class ObservableSubScribeOn<T> extends Observable<T> {
    private Observable<T> observable;
    private Schedule schedule;

    public ObservableSubScribeOn(Observable<T> observable, Schedule schedule) {
        this.observable = observable;
        this.schedule = schedule;
    }

    @Override
    public void subscribeActual(Observer<? super T> observer) {
//        Log.e("TAG","ObservableSubScribeOn subscribeActual");
        Schedule.Worker worker = schedule.createWorker();
        final SubScribeOnObserver<T> scribeOnObserver = new SubScribeOnObserver<>(observer,worker);
        scribeOnObserver.onSubscribe(new DisposeImpl());
        observable.subscribe(scribeOnObserver);

    }
    private static class SubScribeOnObserver<T> implements Observer<T>, Dispose, Runnable {

        private Observer<? super T> observer;
        private final Schedule.Worker worker;
        private Dispose dispose;
        private T t;

        public SubScribeOnObserver(Observer<? super T> observer, Schedule.Worker worker) {
            this.observer = observer;
            this.worker = worker;
        }

        @Override
        public void onSubscribe(final Dispose dispose) {
            this.dispose = dispose;
            observer.onSubscribe(dispose);

        }

        @Override
        public void onNext(@NonNull final T t) throws Exception {
            this.t = t;
            worker.schedule(this);


        }

        @Override
        public void onError(@NonNull final Throwable e) {
            observer.onError(e);

        }

        @Override
        public void onComplete() {
            if (!isDispose())
                observer.onComplete();

        }
        @Override
        public void dispose() {
            DisposableHelper.dispose(dispose);
        }

        @Override
        public boolean isDispose() {
            return DisposableHelper.isDispose(dispose);
        }

        @Override
        public void run() {
            if (!isDispose()) {
                try {
                    observer.onNext(t);
                } catch (Exception e) {
                    onError(e);
                }
                onComplete();
            }
        }
    }
}
